package net.bwie.realtime.jtp.dwd.log.lx2.job;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.KafkaUtil;
import net.bwie.realtime.jtp.dwd.log.function.AdjustIsNewProcessFunction;
import net.bwie.realtime.jtp.dwd.log.lx2.function.AppLogSplitProcessFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.checkerframework.checker.units.qual.K;

public class JtpAppLogEtlJob{

    public static void main(String[] args) throws Exception {
        // 创建 环境
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
        // 并行度
        evn.setParallelism(1);

        // 消费 kafka 中的数据
        DataStream<String> kafka = KafkaUtil.consumerKafka(evn, "topic-log");
//        kafka.print();

        // 数据转换
        DataStream<String> data_conversion = data_conversion(kafka);
        data_conversion.print();
        // 将数据保存到kafka中

        KafkaUtil.producerKafka(data_conversion,"topic-page-log");

        // 触发执行
        evn.execute("lx");
    }

    private static DataStream<String> data_conversion(DataStream<String>  stream) {
        // 1.清洗数据
        DataStream<String> data_cleaning = data_cleaning(stream);

        //2. 添加状态新老访客状态
        DataStream<String> data_status = data_status(data_cleaning);

        //3. 数据分流
        DataStream<String> data_diversion = data_diversion(data_status);


        return data_diversion;
    }




    // 1.清洗数据
    private static DataStream<String> data_cleaning(DataStream<String> stream) {
        // 创建 脏数据的侧边流
        OutputTag<String> display = new OutputTag<String>("displayTag"){};

        SingleOutputStreamOperator<String> process = stream.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String s, Context context, Collector<String> collector) throws Exception {
                        try {
                            // 数据清洗过的数据
                            JSONObject jsonObject = JSON.parseObject(s);
                            collector.collect(s);
                        } catch (Exception e) {
                            context.output(display,s);
                        }
                    }
                }
        );

        SideOutputDataStream<String> sideOutput = process.getSideOutput(display);
        KafkaUtil.producerKafka(sideOutput,"topic-display-log");

        return process;
    }

    //2. 添加状态新老访客状态
    private static DataStream<String> data_status(DataStream<String> stream) {

        KeyedStream<String, String> keyedStream = stream.keyBy(
                new KeySelector<String, String>() {
                    @Override
                    public String getKey(String s) throws Exception {
                        return JSON.parseObject(s).getJSONObject("common").getString("mid");
                    }
                }
        );


        DataStream<String> isNewStream = keyedStream.process(new AdjustIsNewProcessFunction());

        return isNewStream;
    }

    //3. 数据分流
    private static DataStream<String> data_diversion(DataStream<String> stream) {
        OutputTag<String> errorTag = new OutputTag<String>("errorTag"){};
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> actionsTag = new OutputTag<String>("actionsTag"){};
        OutputTag<String> displaysTag = new OutputTag<String>("displaysTag"){};

        SingleOutputStreamOperator<String> process = stream.process(
                new AppLogSplitProcessFunction(
                        errorTag,
                        startTag,
                        displaysTag,
                        actionsTag
                )
        );

        SideOutputDataStream<String> error = process.getSideOutput(errorTag);
        SideOutputDataStream<String> start = process.getSideOutput(startTag);
        SideOutputDataStream<String> actions = process.getSideOutput(actionsTag);
        SideOutputDataStream<String> displays = process.getSideOutput(displaysTag);

        KafkaUtil.producerKafka(error,"topic-error-log");
        KafkaUtil.producerKafka(start,"topic-start-log");
        KafkaUtil.producerKafka(actions,"topic-actions-log");
        KafkaUtil.producerKafka(displays,"topic-displays-log");



        return process;
    }

}
